package com.general.filter;

import static com.general.filter.RpcExceptionExt.INTERFACE_HAS_BEEN_RATE_LIMIT;
import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER;

import io.github.resilience4j.bulkhead.Bulkhead;
import io.github.resilience4j.bulkhead.BulkheadConfig;
import io.github.resilience4j.bulkhead.ThreadPoolBulkhead;
import io.github.resilience4j.bulkhead.ThreadPoolBulkheadConfig;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.Filter;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;


/**
 * @ClassName BulkheadFilter
 * @Desc: BulkheadFilter {@link Bulkhead} and {@link ThreadPoolBulkhead} 集成到filter实现
 * @Author wangxuan
 * @Date: 2021/10/28 13:41
 * @Verson: 1.0
 **/
@Slf4j
@Activate(group = PROVIDER, value = BulkheadConstant.BULK_HEAD_KEY)
public class BulkheadFilter implements Filter{

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String methodName = invocation.getMethodName();
        // 从参数中获取bulkhead的类型
        String bulkheadType = invoker.getUrl()
                .getMethodParameter(methodName, BulkheadConstant.BULK_HEAD_KEY, "semaphore");

        switch (bulkheadType){
            case "semaphore":
                int maxConcurrentCalls = invoker.getUrl()
                        .getMethodParameter(methodName, BulkheadConstant.SEMAPHORE_MAX_CONCURRENT_CALLS, 25);
                int maxWaitDuration = invoker.getUrl()
                        .getMethodParameter(methodName, BulkheadConstant.SEMAPHORE_MAX_WAIT_DURATION, 0);

                // Create a custom configuration for a Bulkhead
                BulkheadConfig config = BulkheadConfig.custom()
                        .maxConcurrentCalls(maxConcurrentCalls)
                        .maxWaitDuration(Duration.ofMillis(maxWaitDuration))
                        .build();

                Bulkhead bulkhead = Bulkhead.of("semaphore", config);

                log.info("bulkhead type is {}, maxConcurrentCalls is {}, maxWaitDuration is {}", bulkheadType, maxConcurrentCalls, maxWaitDuration);
                if (bulkhead.tryAcquirePermission()){
                    log.info("bulkheadType [{}] tryAcquirePermission success", bulkheadType);
                    return invoker.invoke(invocation);
                }
                else{
                    log.error("bulkheadType [{}] tryAcquirePermission failed, url is {}", bulkheadType, invoker.getUrl());
                    throw new RpcExceptionExt(
                            INTERFACE_HAS_BEEN_RATE_LIMIT, "current request rate limit by bulkhead");
                }
            case "threadPool":
                int maxThreadPoolSize = invoker.getUrl()
                        .getMethodParameter(methodName, BulkheadConstant.THREAD_POOL_MAX_THREAD_POOL_SIZE, Runtime.getRuntime()
                                .availableProcessors());

                int coreThreadPoolSize = invoker.getUrl()
                        .getMethodParameter(methodName, BulkheadConstant.THREAD_POOL_CORE_THREAD_POOL_SIZE,
                                Runtime.getRuntime().availableProcessors()-1);

                int queueCapacity = invoker.getUrl()
                        .getMethodParameter(methodName, BulkheadConstant.THREAD_POOL_QUEUE_CAPACITY, 100);

                int keepAliveDuration = invoker.getUrl()
                        .getMethodParameter(methodName, BulkheadConstant.THREAD_POOL_KEEP_ALIVE_DURATION, 20);

                ThreadPoolBulkheadConfig threadPoolBulkheadConfig = ThreadPoolBulkheadConfig.custom()
                        .maxThreadPoolSize(maxThreadPoolSize)
                        .coreThreadPoolSize(coreThreadPoolSize)
                        .queueCapacity(queueCapacity)
                        .keepAliveDuration(Duration.ofMillis(keepAliveDuration))
                        .build();

                log.info("bulkhead type is {}, maxThreadPoolSize is {}, coreThreadPoolSize is {}, queueCapacity is {},"
                        + "keepAliveDuration is {}", bulkheadType, maxThreadPoolSize, coreThreadPoolSize, queueCapacity, keepAliveDuration);
                ThreadPoolBulkhead threadPoolBulkhead = ThreadPoolBulkhead.of("threadPoolBulkhead",
                        threadPoolBulkheadConfig);
                CompletionStage<Result> completionStage = threadPoolBulkhead
                        .executeSupplier(() -> invoker.invoke(invocation));
                CompletableFuture<Result> resultCompletableFuture = completionStage.toCompletableFuture();
                try {
                    return resultCompletableFuture.get();
                } catch (InterruptedException |ExecutionException e) {
                    log.error("bulkheadType [{}] execute exception, url is {}, exception "+e , bulkheadType, invoker.getUrl());
                    throw new RpcExceptionExt(e);
                }
            default:
                log.warn("bulkhead type cannnot match, bulkhead type is {}", bulkheadType);
                return invoker.invoke(invocation);
        }
    }
}
